import { randomBytes } from "crypto";
import { ReadStream } from "fs";
import { ensureDirSync, readFile, readFileSync, rmSync, writeFileSync } from "fs-extra";
import { freemem } from 'os'
import { dirname, join } from "path";
import { logger } from "./logger";

export const GlobalData: Map<string, any> = new Map();
export const getGlobalData = <T>(key: string, init?: () => T) => {
    if (GlobalData.has(key)) return GlobalData.get(key) as T;
    const def = init ? init() : {} as T;
    return def;
}
export const setGlobalData = <T>(key: string, val: T) => {
    return GlobalData.set(key, val)
}
export function useGlobalData<T extends object>(key: string, init?: () => T) {
    const data = getGlobalData<T>(key, init)
    setGlobalData(key, data)
    const setData = (val: Partial<T> | null) => {
        if (val) {
            const obj = {
                ...getGlobalData(key, init),
                ...val
            }
            setGlobalData(key, obj)
        } else {
            GlobalData.delete(key)
        }
    }
    return { data, setData }
}
export interface PromiseResolve<T> {
    (value: T | PromiseLike<T>): void
}
export interface PromiseReject {
    (reason?: any): void;
}
export function usePromise<T>() {
    let resolve: PromiseResolve<T>;
    let reject: PromiseReject;
    const promise = new Promise<T>((r, j) => {
        resolve = r;
        reject = j;
    })
    return { resolve, reject, promise }
}

export class TaskManager {
    tasks: Map<string, Task> = new Map()
    getTaskByName(name: string) {
        return this.tasks.get(name)
    }
    addTask(task: Task) {
        this.tasks.set(task.name, task)
    }
    delTask(task: Task) {
        this.tasks.delete(task.name)
    }
}
export abstract class Task {
    current: number = 0;
    constructor(public name: string, public total: number = 0) { }
    abstract start(multi?: boolean): Promise<void>;
    abstract startFail(): Promise<void>;
    abstract stop(): Promise<void>;
    abstract destroy(): Promise<void>;
}
export interface LoadArgs {
    (): any[];
}
export interface IFunctionTask {
    args: any[] | LoadArgs,
    call: (...args: any[]) => Promise<void>;
    after?: () => void;
}
export class FunctionTask extends Task {
    isStop: boolean = false;
    isEnd: Promise<void>;
    resolve: PromiseResolve<void>;
    reject: PromiseReject;
    hasMore: boolean = true;
    hasDestory: boolean = false;
    failCount: number = 0;
    public fails: IFunctionTask[] = [];
    constructor(name: string, public tasks: IFunctionTask[], public onFinish: (...args: any[]) => Promise<void>) {
        super(name, tasks.length)
        const { resolve, reject, promise: isEnd } = usePromise<void>()
        this.resolve = resolve;
        this.reject = reject;
        this.isEnd = isEnd;
    }
    addFunction(fn: IFunctionTask, hasMore: boolean = true) {
        if (this.hasMore) {
            // save args to local
            const root = process.cwd();
            const filePath = join(root, 'data', 'global', 'data', this.name, `${this.total}.json`)
            ensureDirSync(dirname(filePath))
            writeFileSync(filePath, JSON.stringify(fn.args))
            fn.args = () => {
                return JSON.parse(readFileSync(filePath, 'utf-8'))
            }
            fn.after = () => {
                rmSync(filePath)
            }
            this.tasks.push(fn)
            this.total += 1;
            this.hasMore = hasMore;
        }
    }
    private delay(time: number) {
        const { resolve, promise } = usePromise()
        setTimeout(resolve, time)
        return promise;
    }
    async startFail() {
        if (this.hasDestory) return;
        let task = this.fails.shift()
        while (task) {
            let args: any[] = [];
            if (typeof task.args === 'function') {
                args = task.args();
            } else {
                args = task.args;
            }
            try {
                await task.call(...args)
                task.after && task.after()
                this.current += 1;
                this.failCount -= 1;
                await this.onFinish(null, ...args)
                task = this.tasks.shift()
            } catch (e) {
                console.error(e)
                this.fails.push(task)
                await this.onFinish(e, ...args)
            }
        }
    }
    async start(multi: boolean = false): Promise<void> {
        if (multi) return this.startMulti()
        if (this.hasDestory) return;
        let task = this.tasks.shift()
        while (!this.isStop && task) {
            let args: any[] = [];
            if (typeof task.args === 'function') {
                args = task.args();
            } else {
                args = task.args;
            }
            try {
                await task.call(...args)
                task.after && task.after()
                this.current += 1;
                await this.onFinish(null, ...args)
                task = this.tasks.shift()
            } catch (e) {
                this.failCount += 1;
                this.fails.push(task)
                await this.onFinish(e, ...args)
            }
        }
        if (!this.hasMore && this.tasks.length <= 0) {
            this.resolve()
            return;
        }
        if (this.isStop) {
            return;
        }
        await this.start(multi);
    }
    task(len: number = 0) {
        if (len) {
            return this.tasks.splice(0, len)
        } else {
            const tasks = this.tasks;
            this.tasks = [];
            return tasks;
        }
    }
    length: number = 0;
    maxLength: number = 50;
    async startMulti() {
        if (this.hasDestory) return;
        if (this.length < this.maxLength) {
            const tasks = this.task(this.maxLength - this.length);
            let task = tasks.pop();
            while (task && this.length < this.maxLength) {
                this.length += 1;
                let args: any[] = [];
                if (typeof task.args === 'function') {
                    args = task.args();
                } else {
                    args = task.args;
                }
                const after = task.after;
                task.call(...args).then(async () => {
                    this.current += 1;
                    this.length -= 1;
                    await this.whenTaskFinish()
                    logger.info(`current ${this.current}-${this.length} task is success`)
                    after && await after()
                    await this.onFinish(null, ...args)
                }).catch(async e => {
                    this.failCount += 1;
                    this.current += 1;
                    this.length -= 1;
                    await this.whenTaskFinish()
                    logger.info(`current ${this.current} ${this.length} task is fail ${e.message}`)
                    await this.onFinish(e, ...args)
                })
                task = tasks.pop();
            }
        }
        logger.info(`${this.hasMore ? '还有更多呢' : '没有更多了'}, 任务堆长度${this.tasks.length}`)
        if (!this.hasMore && this.tasks.length <= 0) {
            return;
        }
        await this.delay(1000)
        this.startMulti();
    }
    async whenTaskFinish() {
        if (this.current === this.total) {
            this.resolve()
        }
    }
    async stop(): Promise<void> {
        if (this.destroy) return;
        this.isStop = true;
    }
    async destroy(): Promise<void> {
        this.tasks = [];
        this.hasDestory = true;
    }
}
export interface IReadStreamTask {
    name: string,
    total: number,
    stream: ReadStream
}
export abstract class ReadStreamTask extends Task {
    stream: ReadStream;
    isFinish: Promise<void>;
    resolve: PromiseResolve<void>;
    hasDestory: boolean = false;
    isStop: boolean = false;
    constructor(options: IReadStreamTask) {
        super(options.name, options.total)
        const { resolve, reject, promise } = usePromise<void>()
        this.resolve = resolve;
        this.stream = options.stream;
        this.stream.pause();
        this.stream.on('data', (chunk) => {
            this.current += chunk.length;
            this.handle(chunk);
        });
        this.stream.on('close', () => this.onEnd())
        this.stream.on('end', () => this.onEnd())
        this.stream.on('error', (err) => reject(err))
        this.isFinish = promise;
    }
    abstract handle(chunk: string | Buffer): Promise<void>;
    async start(multi: boolean = true): Promise<void> {
        if (this.hasDestory) return;
        if (this.isStop) {
            return;
        }
        this.stream.resume();
        logger.info(`resume ......`)
        this.isStop = false;
    }
    async stop(): Promise<void> {
        if (this.hasDestory) return;
        if (this.isStop) {
            return;
        }
        this.stream.pause();
        logger.info(`pause ......`)
        this.isStop = true;
    }
    abstract onEnd(): Promise<void>;
    async destroy() {
        this.hasDestory = true;
        this.stream.destroy()
    }
}

export abstract class CsvReadStreamTask extends ReadStreamTask {
    private str: string = ``;
    private hasData: boolean = false;
    private headers: string[] = [];
    private rowData: string[][] = [];
    private id: string = ``;
    private task: FunctionTask;
    public isEnd: boolean = false;
    // public maxFunction: number = 100;
    public isStoped: boolean = false
    public extra: any = {}
    constructor(private hasHeader: boolean, private minLength: number, private onFinish: (...args: any[]) => Promise<void>, options: IReadStreamTask) {
        super(options)
        this.id = `${Date.now()}${randomBytes(12).toString('base64')}`
        this.task = new FunctionTask(this.id, [], onFinish)
        this.task.isEnd.then(async () => {
            this.isEnd = true;
            await onFinish(null);
            this.resolve()
        })
    }
    toPayload() {
        return {
            total: this.total,
            current: this.current,
            isEnd: this.isEnd,
            taskCount: this.task.current,
            taskTotal: this.task.total,
            failCount: this.task.failCount,
            freemem: freemem(),
            name: this.name,
            ...this.extra,
        }
    }
    async handle(buf: string | Buffer): Promise<void> {
        const data = buf.toString('utf-8')
        this.str += data;
        const list = this.str.split('\n')
        const end = list.pop();
        this.str = end;
        const listData = list.map(it => it.replace('\r', '')).map(it => it.split('|'))
        if (!this.hasData) {
            if (this.hasHeader) {
                this.headers = listData.shift();
            }
            this.hasData = true;
        }
        this.rowData.push(...listData)
        if (this.rowData.length >= this.minLength) {
            const rows = this.rowData;
            this.rowData = [];
            this.handleRows(rows, true)
        }
    }
    async handleRows(rows: string[][], hasMore: boolean): Promise<void> {
        this.task.addFunction({
            args: [rows],
            call: this.onRows.bind(this)
        }, hasMore)
    }
    abstract onRows(rows: string[][]): Promise<void>;
    abstract clear(): Promise<void>;
    async start(multi: boolean = true): Promise<void> {
        super.start(multi);
        this.task.start(multi);
    }
    async startFail() {
        this.task.startFail()
    }
    async stop(): Promise<void> {
        super.stop();
        this.task.stop();
    }
    async destroy(): Promise<void> {
        super.destroy()
        this.task.destroy();
    }
    async onEnd(): Promise<void> {
        const list = this.str.split('\n')
        const listData = list.filter(it => !!it).map(it => it.replace('\r', '')).map(it => it.split('|'))
        this.rowData.push(...listData)
        const rows = this.rowData;
        this.rowData = [];
        this.handleRows(rows, false)
        this.resolve()
    }
}